package cn.caplike.demo.rabbitmq.springboot.consumer;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.Objects;

@Slf4j
@Component
public class MessageConsumer {

    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = "queue-1", durable = "false", autoDelete = "true"),
                    exchange = @Exchange(value = "exchange-1", durable = "false", autoDelete = "true", type = "topic", ignoreDeclarationExceptions = "true"),
                    key = "springboot.*"
            ),
            ackMode = "MANUAL"
    )
    @RabbitHandler
    public void onMessage(Message<?> message, Channel channel) throws Exception {
        long deliveryTag = (long) Objects.requireNonNull(
                message.getHeaders().get(AmqpHeaders.DELIVERY_TAG),
                String.format("%s 不能为空!", AmqpHeaders.DELIVERY_TAG)
        );

        channel.basicNack(deliveryTag, true, false);

//        Thread.sleep(10 * 1000);
//        log.error("--------------------------------------");
//        log.error("消费端Payload: " + message.getPayload());
//        // 手工ACK
//        // channel.basicAck(Objects.requireNonNull(deliveryTag, "deliveryTag cannot be null!"), false);
//        log.error("要 nack 了 !!!");
//        channel.basicNack(deliveryTag, false, false);
    }

//    /**
//     * Description: 用配置初始化<br>
//     * Details:
//     * <dl>
//     *     <dd>spring.rabbitmq.listener.order.queue.name=queue-2</dd>
//     *     <dd>spring.rabbitmq.listener.order.queue.durable=true</dd>
//     *     <dd>spring.rabbitmq.listener.order.exchange.name=exchange-1</dd>
//     *     <dd>spring.rabbitmq.listener.order.exchange.durable=true</dd>
//     *     <dd>spring.rabbitmq.listener.order.exchange.type=topic</dd>
//     *     <dd>spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true</dd>
//     *     <dd>spring.rabbitmq.listener.order.key=springboot.*</dd>
//     * </dl>
//     */
//    @RabbitListener(
//            bindings = @QueueBinding(
//                    value = @Queue(value = "${spring.rabbitmq.listener.order.queue.name}",
//                            durable = "${spring.rabbitmq.listener.order.queue.durable}"),
//                    exchange = @Exchange(value = "${spring.rabbitmq.listener.order.exchange.name}",
//                            durable = "${spring.rabbitmq.listener.order.exchange.durable}",
//                            type = "${spring.rabbitmq.listener.order.exchange.type}",
//                            ignoreDeclarationExceptions = "${spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions}"),
//                    key = "${spring.rabbitmq.listener.order.key}"
//            )
//    )
//    @RabbitHandler
//    public void onOrderMessage(@Payload Order order,
//                               Channel channel,
//                               @Headers Map<String, Object> headers) throws Exception {
//        System.err.println("--------------------------------------");
//        System.err.println("消费端order: " + order.getId());
//        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
//        // 手工ACK
//        channel.basicAck(deliveryTag, false);
//    }

    @Bean
    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return factory;
    }

}
